#include "my_aio.h"
#include "my_worker_processor.h"

#define MY_AIO_LOOP_STOP_ERRNO              (11111)

char *my_aio_thread_tag = "aio_thread_tag";


static inline void
my_aio_loop_executer(my_aio_loop_data_t *loop_data, my_bool_t *can_stop)
{
    my_queue_t              *q = NULL;
    my_dispatch_event_t     *dispatch_event;
    my_aio_request_t        *aio_req;

    my_mutex_lock(&loop_data->mutex);

    if (!my_queue_empty(&loop_data->extern_queue)) {
        q = my_queue_last(&loop_data->extern_queue);
        my_queue_remove(q);
    } else {
        if (loop_data->stop) {
            *can_stop = MY_TRUE;
        } else {
            my_cond_wait(&loop_data->cond, &loop_data->mutex);
        }
    }

    my_mutex_unlock(&loop_data->mutex);

    if (!q) {
        return;
    }

    dispatch_event = my_queue_data(q, my_dispatch_event_t, queue);
    aio_req = (my_aio_request_t *)dispatch_event->save_data;

    switch (aio_req->op) {
    case MY_AIO_READ:
        aio_req->res =
            my_syscall_pread(aio_req->fd, aio_req->buf, aio_req->nbytes, aio_req->offset);
        break;
    case MY_AIO_WRITE:
        aio_req->res =
            my_syscall_pwrite(aio_req->fd, aio_req->buf, aio_req->nbytes, aio_req->offset);
        break;
    default:
        break;
    }

    my_atomic_int64_add_fetch(&loop_data->tasks, -1);
    aio_req->done_time = my_usec_2_msec(get_cache_time());

    my_event_loop_post_event(dispatch_event->setup_thread, dispatch_event);
}


static inline void *
my_aio_loop_runner(my_thread_base_t *thread_base)
{
    my_aio_loop_data_t *loop_data = thread_base->data;

    my_bool_t can_stop          = MY_FALSE;

    for ( ;; ) {
        my_aio_loop_executer(loop_data, &can_stop);
        if (can_stop) {
            break;
        }
    }

    return NULL;
}


void
my_aio_manager_start(my_aio_manager_t *aio_manager,
    int n_read_threads, int n_write_threads)
{
    if (n_read_threads <= 0 || n_write_threads <= 0) {
        MY_DEF_LOG_FATAL("n_read_threads or n_write_threads must be greate than 0");
    }

    bzero(aio_manager, sizeof(my_aio_manager_t));
    aio_manager->n_read_threads = n_read_threads;
    aio_manager->n_write_threads = n_write_threads;

    my_aio_loop_data_init(&aio_manager->read_loop_data);
    my_aio_loop_data_init(&aio_manager->write_loop_data);

    aio_manager->read_threads =
        my_thread_calloc(sizeof(my_thread_base_t) * (size_t)n_read_threads);
    aio_manager->write_threads =
        my_thread_calloc(sizeof(my_thread_base_t) * (size_t)n_write_threads);

    for (int i = 0; i < n_read_threads; i++) {
        my_thread_base_start(
            my_aio_thread_tag,
            aio_manager->read_threads + i,
            my_aio_loop_runner,
            &aio_manager->read_loop_data,
            0, 0);
    }

    for (int i = 0; i < n_write_threads; i++) {
        my_thread_base_start(
            my_aio_thread_tag,
            aio_manager->write_threads + i,
            my_aio_loop_runner,
            &aio_manager->write_loop_data,
            0, 0);
    }
}


void
my_aio_manager_stop(my_aio_manager_t *aio_manager)
{
    my_mutex_lock(&aio_manager->read_loop_data.mutex);
    aio_manager->read_loop_data.stop = MY_TRUE;
    my_cond_broadcast(&aio_manager->read_loop_data.cond);
    my_mutex_unlock(&aio_manager->read_loop_data.mutex);

    my_mutex_lock(&aio_manager->write_loop_data.mutex);
    aio_manager->write_loop_data.stop = MY_TRUE;
    my_cond_broadcast(&aio_manager->write_loop_data.cond);
    my_mutex_unlock(&aio_manager->write_loop_data.mutex);

    for (int i = 0; i < aio_manager->n_read_threads; i++) {
        my_thread_base_join(aio_manager->read_threads + i, NULL);
    }

    for (int i = 0; i < aio_manager->n_write_threads; i++) {
        my_thread_base_join(aio_manager->write_threads + i, NULL);
    }

    my_thread_free(aio_manager->read_threads);
    my_thread_free(aio_manager->write_threads);
}


// the following interfaces can only be called in the worker thread.
void
my_aio_post_event(my_aio_manager_t *aio_manager,
    my_aio_request_t *req)
{
    assert(current_thread_is_worker_thread());

    my_aio_loop_data_t *loop_data =
        &aio_manager->read_loop_data;
    if (req->op == MY_AIO_WRITE || req->op == MY_AIO_WRITE_PING) {
        loop_data = &aio_manager->write_loop_data;
    }

    req->dispatch_event.setup_thread = current_thread_base();
    req->setup_time = my_usec_2_msec(get_cache_time());

    my_atomic_int64_add_fetch(&loop_data->tasks, 1);

    my_bool_t loop_stop = MY_FALSE;

    my_mutex_lock(&loop_data->mutex);
    if (!loop_data->stop) {
        my_queue_insert_head(&loop_data->extern_queue, &req->dispatch_event.queue);
        my_cond_signal(&loop_data->cond);
    } else {
        loop_stop = MY_TRUE;
    }
    my_mutex_unlock(&loop_data->mutex);


    if (!loop_stop) {
        return;
    }

    req->res.res = 0;
    req->res.eno = MY_AIO_LOOP_STOP_ERRNO;
    my_atomic_int64_add_fetch(&loop_data->tasks, -1);
    req->done_time = my_usec_2_msec(get_cache_time());
    my_event_loop_post_event(req->dispatch_event.setup_thread, &req->dispatch_event);
}
